package com.zlm.app;

import com.zlm.bean.OrderEvent;
import com.zlm.bean.OrderResult;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.net.URL;

/**
 * Author: Harbour
 * Date: 2021-05-30 16:33
 * Desc: 使用ProcessFunction来处理订单支付超时逻辑
 */
public class OrderTimeoutAPP {
    public static void main(String[] args) throws Exception {

        // step 1: get Env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // step 2: get data source
        URL resource = OrderTimeoutAPP.class.getResource("/OrderLog.csv");
        DataStreamSource<String> inputDataStream = env.readTextFile(resource.getPath());

        // step 3: transform data source
        SingleOutputStreamOperator<OrderEvent> orderEventDataStream = inputDataStream.map(new MapFunction<String, OrderEvent>() {
            @Override
            public OrderEvent map(String value) throws Exception {
                String[] fields = value.split(",");
                return new OrderEvent(
                        Long.parseLong(fields[0]),
                        fields[2],
                        fields[3],
                        Long.parseLong(fields[3])
                );
            }
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
            @Override
            public long extractAscendingTimestamp(OrderEvent element) {
                return element.getTimestamp();
            }
        });

        // step 4. use processFunction to process dataStream
        SingleOutputStreamOperator<OrderResult> orderResultDataStream = orderEventDataStream.keyBy(OrderEvent::getOrderId)
                .process(new OrderTimeoutProcessFunction());

        // step 5: get result
        orderResultDataStream.print();
        orderResultDataStream.getSideOutput(orderTimeoutOutputTag).print();

        env.execute("Order timeout Job");
    }

    private final static OutputTag<OrderResult> orderTimeoutOutputTag = new OutputTag<OrderResult>("orderTimeout") {};

    private static class OrderTimeoutProcessFunction extends KeyedProcessFunction<Long, OrderEvent, OrderResult> {

        private ValueState<Boolean> isPayed;
        private ValueState<Boolean> isCreated;
        private ValueState<Long> timer;

        @Override
        public void open(Configuration parameters) throws Exception {
            // define and initial some status
            isPayed = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isPayed", Boolean.class, false));
            isCreated = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isCreated", Boolean.class, false));
            timer = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer", Long.class));
        }

        @Override
        public void processElement(OrderEvent value, Context ctx, Collector<OrderResult> out) throws Exception {
            String eventType = value.getEventType();

            if ("create".equalsIgnoreCase(eventType)) {
                if (isPayed.value()) {
                    // case 1 已创建 已支付, 则清空定时器
                    ctx.timerService().deleteEventTimeTimer(timer.value());
                    isPayed.clear();
                    timer.clear();
                    out.collect(
                        new OrderResult(
                            value.getOrderId(),
                            "payed successfully"
                        )
                    );

                } else {
                    // case 2 已创建 未支付, 则开启一个定时器
                    long expirationTime = (value.getTimestamp() + 15 * 60) * 1000L;
                    ctx.timerService().registerEventTimeTimer(expirationTime);
                    isCreated.update(true);
                    timer.update(expirationTime);
                }
            } else if ("pay".equalsIgnoreCase(eventType)) {
                if (isCreated.value()) {
                    // case 3 订单已创建
                    if (value.getTimestamp() * 1000L < timer.value()) {
                        out.collect(
                            new OrderResult(
                                value.getOrderId(),
                                "payed successfully"
                            )
                        );
                    } else {
                        ctx.output(
                            orderTimeoutOutputTag,
                            new OrderResult(
                                value.getOrderId(),
                                "payed but already timeout"
                            )
                        );
                    }
                    isCreated.clear();
                    timer.clear();
                    ctx.timerService().deleteEventTimeTimer(timer.value());

                } else {
                    // case 4 订单未创建 ？
                    ctx.timerService().registerEventTimeTimer(value.getTimestamp() * 1000L);
                    isPayed.update(true);
                    timer.update(value.getTimestamp() * 1000L);
                }
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderResult> out) throws Exception {
            if (isPayed.value()) {
                ctx.output(
                    orderTimeoutOutputTag,
                    new OrderResult(
                        ctx.getCurrentKey(),
                        "already payed but not found create log"
                    )
                );
            } else {
                ctx.output(
                    orderTimeoutOutputTag,
                    new OrderResult(
                        ctx.getCurrentKey(),
                        "payed timeout"
                    )
                );
            }
            isCreated.clear();
            isPayed.clear();
            timer.clear();
        }
    }
}
